package com.practice.gmall.realtime.app.dwd;

import com.alibaba.fastjson.*;
import com.practice.gmall.realtime.app.BaseApp;
import com.practice.gmall.realtime.common.Constant;
import com.practice.gmall.realtime.util.FlinkSinkUtil;
import com.practice.gmall.realtime.util.MyTimeUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.HashMap;
import java.util.Objects;

@Slf4j
public class DWD_01_TradeBaseLog extends BaseApp {
    private static final String ACTION = "action";
    private static final String START = "start";
    private static final String DISPLAY = "display";
    private static final String ERR = "err";
    private static final String PAGE = "page";

    // 将dwd
    public static void main(String[] args) {
        new DWD_01_TradeBaseLog().init(3334, "DWD_01_TradeBaseLog", 2, Constant.KAFKA_BROKERS, Constant.TOPIC_ODS_LOG);
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 1. etl
        SingleOutputStreamOperator<JSONObject> etledStream = etl(stream);
        // 2. correct is_new tag, use keyed state API
        SingleOutputStreamOperator<JSONObject> correctedStream = correctIsNewTag(etledStream);
        // 3. classify 5 types of log data (error,start,display,action,pages), use side output API
        HashMap<String, DataStream<JSONObject>> streams = classifyStream(correctedStream);
        //streams.get(ACTION).print();
        // 4. write to kafka
        streams.get(ACTION).map(JSONAware::toJSONString).addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ACTION));
        streams.get(START).map(JSONAware::toJSONString).addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_START));
        streams.get(DISPLAY).map(JSONAware::toJSONString).addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_DISPLAY));
        streams.get(ERR).map(JSONAware::toJSONString).addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ERR));
        streams.get(PAGE).map(JSONAware::toJSONString).addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_PAGE));
    }

    private HashMap<String, DataStream<JSONObject>> classifyStream(SingleOutputStreamOperator<JSONObject> stream) {
        // todo 为什么需要加大括号
        OutputTag<JSONObject> errorTag = new OutputTag<JSONObject>("errorTag"){};
        //OutputTag<JSONObject> startTag = new OutputTag<JSONObject>("startTag"){};
        OutputTag<JSONObject> displayTag = new OutputTag<JSONObject>("displayTag"){};
        OutputTag<JSONObject> actionTag = new OutputTag<JSONObject>("actionTag"){};
        OutputTag<JSONObject> pageTag = new OutputTag<JSONObject>("pageTag"){};

        SingleOutputStreamOperator<JSONObject> startStream = stream.process(new ProcessFunction<JSONObject, JSONObject>() {
            @Override
            public void processElement(JSONObject originObj,
                                       Context ctx,
                                       Collector<JSONObject> out) throws Exception {
                JSONObject common = originObj.getJSONObject("common");
                JSONObject page = originObj.getJSONObject("page");
                JSONObject start = originObj.getJSONObject("start");
                Long ts = originObj.getLong("ts");

                if (start != null) {
                    out.collect(originObj);
                }

                JSONArray displays = originObj.getJSONArray("displays");

                if (displays != null) {
                    for (int i = 0; i < displays.size(); i++) {
                        JSONObject display = displays.getJSONObject(i);
                        display.putAll(common);
                        display.putAll(page);
                        display.put("ts", ts);
                        ctx.output(displayTag, display);
                    }
                    originObj.remove("displays");
                }

                JSONArray actions = originObj.getJSONArray("actions");

                if (actions != null) {
                    for (int i = 0; i < actions.size(); i++) {
                        JSONObject action = actions.getJSONObject(i);
                        action.putAll(common);
                        action.putAll(page);
                        action.put("ts", ts);
                        ctx.output(actionTag, action);
                    }
                    originObj.remove("actions");
                }

                // output to err
                JSONObject err = originObj.getJSONObject("err");
                if (err != null) {
                    ctx.output(errorTag, originObj);//todo 为什么这里是把originalObject传出去？
                    originObj.remove("err");
                }

                // output to page
                if (page != null) {
                    ctx.output(pageTag, originObj);
                }


            }
        });

        DataStream<JSONObject> actionStream = startStream.getSideOutput(actionTag);
        DataStream<JSONObject> displayStream = startStream.getSideOutput(displayTag);
        DataStream<JSONObject> errStream = startStream.getSideOutput(errorTag);
        DataStream<JSONObject> pageStream = startStream.getSideOutput(pageTag);

        HashMap<String, DataStream<JSONObject>> streams = new HashMap<>();

        streams.put(START, startStream);
        streams.put(ACTION, actionStream);
        streams.put(DISPLAY, displayStream);
        streams.put(ERR, errStream);
        streams.put(PAGE, pageStream);
        return streams;
    }


    private SingleOutputStreamOperator<JSONObject> correctIsNewTag(SingleOutputStreamOperator<JSONObject> stream) {

        return stream.keyBy(obj -> obj.getJSONObject("common").getString("mid"))
                .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    private ValueState<String> signUpDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        signUpDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("firstDate", String.class));
                    }

                    @Override
                    public void processElement(JSONObject originObj,
                                               Context ctx,
                                               Collector<JSONObject> out) throws Exception {

                        JSONObject common = originObj.getJSONObject("common");
                        String isNew = common.getString("is_new");//todo  bug1
                        Long ts = originObj.getLong("ts");
                        String signUpdate = signUpDateState.value();
                        String today = MyTimeUtils.formatMillisToDateString(ts);

                        // todo logic is important
                        if (Objects.equals(isNew, "1")) {
                            if (signUpdate == null) {
                                signUpDateState.update(today);
                            } else if (!Objects.equals(signUpdate, today)) {
                                common.put("is_new",0);
                                originObj.put("common",common);
                            }
                        } else {// isNew = null or isNew = "0"
                            if (signUpdate == null) {
                                signUpDateState.update(MyTimeUtils.formatMillisToDateString(ts - 24 * 60 * 60 * 1000));
                            }
                        }
                        out.collect(originObj);
                    }
                });


    }

    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {
        return stream.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String str, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject obj = JSON.parseObject(str);
                    out.collect(obj);
                } catch (JSONException e) {
                    log.warn("Can not cast to JSONObject : " + str);
                }
            }
        });
    }
}
